package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.beans.OrderWide;
import com.atguigu.gmall.realtime.beans.PaymentInfo;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * Author: Felix
 * Date: 2021/8/10
 * Desc: 支付宽表准备
 */
public class PaymentWideApp01 {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点设置 (略)

        //TODO 3.从Kafka中读取数据
        //3.1 声明消费主题以及消费者组
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String groupId = "payment_wide_app_group";

        //3.2 获取消费者对象
        //支付
        FlinkKafkaConsumer<String> paymentInfoSource = MyKafkaUtil.getKafkaSource(paymentInfoSourceTopic, groupId);
        //订单宽表
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId);

        //3.3 读取数据  封装流
        //支付
        DataStreamSource<String> paymentInfoStrDS = env.addSource(paymentInfoSource);
        //订单宽表
        DataStreamSource<String> orderWideStrDS = env.addSource(orderWideSource);

        //TODO 4.对数据类型进行转换  String-> 实体对象
        //支付
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = paymentInfoStrDS.map(jsonStr ->
                JSON.parseObject(jsonStr, PaymentInfo.class));
        //订单宽表
        SingleOutputStreamOperator<OrderWide> orderWideDS = orderWideStrDS.map(jsonStr ->
                JSON.parseObject(jsonStr, OrderWide.class));

        paymentInfoDS.print(">>>>");
        orderWideDS.print("#####");
        env.execute();
    }
}


















